package com.zk.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class HBaseDataReader implements AutoCloseable {
    private final Connection connection;
    private final Table countryTable;
    private final Table platformTable;
    private final Table deviceTable;
    private final Table agentTable;

    public HBaseDataReader() throws IOException {
        //创建Hbase连接
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "localhost:2183");
        connection = ConnectionFactory.createConnection(config);
        // 获取各个聚合结果表的Table对象
        countryTable = connection.getTable(TableName.valueOf("country"));
        platformTable = connection.getTable(TableName.valueOf("platform"));
        deviceTable = connection.getTable(TableName.valueOf("device"));
        agentTable = connection.getTable(TableName.valueOf("agent"));
    }

    // 国家总点击量查询
    public Map<String, Long> getCountrySums(String date) throws IOException {return getAggResults(countryTable, date);}
    // 系统平均点击量查询
    public Map<String, Double> getPlatformAvgs(String date) throws IOException {return getAggResults(platformTable, date);}
    // 设备最大点击量查询
    public Map<String, Integer> getDeviceMaxs(String date) throws IOException {return getAggResults(deviceTable, date);}
    // 浏览器总点击量查询
    public Map<String, Long> getAgentSums(String date) throws IOException {return getAggResults(agentTable, date);}

    // 通用聚合查询方法
    private <T> Map<String, T> getAggResults(Table table, String date) throws IOException {
        Map<String, T> results = new HashMap<>();// 创建结果集与扫描对象
        Scan scan = new Scan();
        scan.setRowPrefixFilter(Bytes.toBytes(date));// 设置行键前缀过滤器，只扫描指定日期的数据

        try (ResultScanner scanner = table.getScanner(scan)) {
            for (Result result : scanner) {
                String rowKey = Bytes.toString(result.getRow());
                String value = rowKey.substring(date.length() + 1); // 移除日期前缀和分隔符
                // 根据表类型确定列名和值提取器
                if (table == countryTable || table == agentTable) {
                    byte[] colValue = result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("sum_hits"));
                    if (colValue != null) {
                        results.put(value, (T) Long.valueOf(Bytes.toLong(colValue)));//转化为Long
                    }
                }
                else if (table == platformTable) {
                    byte[] colValue = result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("avg_hits"));
                    if (colValue != null) {
                        results.put(value, (T) Double.valueOf(Bytes.toDouble(colValue)));//转化为Double
                    }
                }
                else if (table == deviceTable) {
                    byte[] colValue = result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("max_hits"));
                    if (colValue != null) {
                        results.put(value, (T) Integer.valueOf(Bytes.toInt(colValue)));//转化为Int
                    }
                }
            }
        }
        return results;
    }

    @Override
    public void close() throws IOException {
        countryTable.close();
        platformTable.close();
        deviceTable.close();
        agentTable.close();
        connection.close();
    }

    /****************** 使用示例 ******************/
    public static void main(String[] args) {
        try (HBaseDataReader reader = new HBaseDataReader()) {
            String date = "20250619";

            System.out.println("===== 国家/地区总点击量 =====");
            reader.getCountrySums(date).forEach(
                    (country, sum) -> System.out.printf("%-15s: %,d%n", country, sum));

            System.out.println("\n===== 系统平均点击量 =====");
            reader.getPlatformAvgs(date).forEach(
                    (platform, avg) -> System.out.printf("%-10s: %.2f%n", platform, avg));

            System.out.println("\n===== 设备最大点击量 =====");
            reader.getDeviceMaxs(date).forEach(
                    (device, max) -> System.out.printf("%-15s: %,d%n", device, max));

            System.out.println("\n===== 浏览器总点击量 =====");
            reader.getAgentSums(date).forEach(
                    (agent, sum) -> System.out.printf("%-20s: %,d%n", agent, sum));
        } catch (IOException e) {
            System.err.println("HBase操作失败: " + e.getMessage());
        }
    }
}